package com.ylkj.process;

import com.alibaba.fastjson2.JSONObject;
import com.hwl.datahub.SchemaVo;
import com.ylkj.bus.MessageContent;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Author: wengzx
 * @Date: 2025/3/12
 * @Desc:
 */
@Slf4j
public class UnpackConvertProcess extends ProcessFunction<List<SchemaVo>, MessageContent>{
    private static final String TOPIC = "topic";
    private static final String DATA = "data";
    private static final String CLIENTID = "clientid";

    /**
     * HEADERS
     */
    public static final String HEADERS = "headers";

    @Override
    public void processElement(List<SchemaVo> values, ProcessFunction<List<SchemaVo>, MessageContent>.Context ctx, Collector<MessageContent> out) throws Exception {
        if(values == null || values.isEmpty()){
            return;
        }
        while (!values.isEmpty()){
            SchemaVo remove = values.remove(values.size() - 1);
            MessageContent messageContent = convert(remove);
            out.collect(messageContent);
        }
    }

    private MessageContent convert(SchemaVo schemaVo){
        String topic = schemaVo.get(TOPIC);
        String data = schemaVo.get(DATA);
        String clientId = schemaVo.get(CLIENTID);
        if(StringUtils.isBlank(topic) || StringUtils.isBlank(data) || StringUtils.isBlank(clientId)){
            log.debug("原始报文数据topic:{},data:{},clientId:{}为空",topic,data,clientId);
        }
        MessageContent mqttMsg = new MessageContent(clientId, data,topic);
        mqttMsg.setHeaders(buildHeaders(schemaVo.getContent()));
        return mqttMsg;
    }

    private Map<String, String> buildHeaders(JSONObject content) {
        if(content.containsKey(HEADERS)){
            String field = content.getString(HEADERS);
            JSONObject jsonObject = JSONObject.parseObject(field);
            Map<String, String> headers = new HashMap<>(jsonObject.size());
            for (String key : jsonObject.keySet()) {
                headers.put(key, jsonObject.getString(key));
            }
            return headers;
        }
        return Collections.emptyMap();
    }
}
